package edu.xhu.behavior.component;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import edu.xhu.api.id.IdClient;
import edu.xhu.behavior.mapper.BiliUserDanmuMapper;
import edu.xhu.behavior.util.IdGen;
import edu.xhu.common.exception.CustomException;
import edu.xhu.model.behavior.enums.DanMuState;
import edu.xhu.model.behavior.pojos.BiliUserDanmu;
import edu.xhu.model.biliuser.pojos.BiliUser;
import edu.xhu.model.common.constants.kafka.topic.DanmuTopic;
import edu.xhu.model.common.enums.AppHttpCodeEnum;
import edu.xhu.model.common.enums.IsFlagEnum;
import edu.xhu.utils.thread.BiliUserThreadLocalUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.util.Date;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Component
@RequiredArgsConstructor
@ServerEndpoint(value = "/ws/danmu/{vid}")
public class DanmuWebSocketServer {

    private final KafkaTemplate<String, String> kafkaTemplate;

    private final IdClient idClient;

    // 对每个视频存储该视频下的session集合
    private static final Map<String, Set<Session>> videoConnectionMap = new ConcurrentHashMap<>();

    /**
     * 连接建立时触发，记录session到map
     *
     * @param session 会话
     * @param vid     视频的ID
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("vid") String vid) {
        if (videoConnectionMap.get(vid) == null) {
            Set<Session> set = new HashSet<>();
            set.add(session);
            videoConnectionMap.put(vid, set);
        } else {
            videoConnectionMap.get(vid).add(session);
        }
        sendMessage(vid, "当前观看人数" + videoConnectionMap.get(vid).size());
    }

    /**
     * 收到消息时触发，记录到数据库并转发到对应的全部连接
     *
     * @param session 当前会话
     * @param message 信息体（包含"token"、"vid"、"data"字段）
     * @param vid     视频ID
     */
    @OnMessage
    @Transactional
    public void onMessage(Session session, String message, @PathParam("vid") String vid) {
        BiliUser user = BiliUserThreadLocalUtil.getUser();
        if (session == null || message == null || vid == null || user == null || user.getUserId() == null) {
            log.info("参数错误");
            return;
        }
        try {
            JSONObject msg = JSON.parseObject(message);
            // 持久化
            JSONObject data = msg.getJSONObject("data");
            BiliUserDanmu danmu = new BiliUserDanmu();
            danmu.setDmId(IdGen.getBehaviorId(idClient))
                    .setVideoId(Long.parseLong(vid))
                    .setUserId(user.getUserId())
                    .setContent(data.getString("content"))
                    .setFontSize(data.getInteger("fontsize"))
                    .setMode(data.getInteger("mode").toString())
                    .setColor(data.getString("color"))
                    .setTimePoint(data.getInteger("timePoint"))
                    .setCreateTime(new Date())
                    .setState(DanMuState.PASS.getCode())
                    .setIsShield(IsFlagEnum.NO.getCode());
            //插入弹幕,以及视频弹幕+1,走kafka
            kafkaTemplate.send(DanmuTopic.BILL_Danmu_TOPIC, JSON.toJSONString(danmu));
            // 广播弹幕
            String dmJson = JSON.toJSONString(danmu);
            sendMessage(vid, dmJson);
        } catch (Exception e) {
            log.info("弹幕持久化广播失败~{}", e.toString());
        }
    }

    /**
     * 连接关闭时执行
     *
     * @param session 当前会话
     * @param vid     视频ID
     */
    @OnClose
    public void onClose(Session session, @PathParam("vid") String vid) {
        // 从缓存中移除连接记录
        videoConnectionMap.get(vid).remove(session);
        if (videoConnectionMap.get(vid).isEmpty()) {
            // 如果没人了就直接移除这个视频
            videoConnectionMap.remove(vid);
        } else {
            // 否则更新在线人数
            try {
                sendMessage(vid, "当前观看人数" + videoConnectionMap.get(vid).size());
            } catch (Exception e) {
                log.info("更新在线人数失败！{}", e.toString());
            }
        }
    }

    @OnError
    public void onError(Throwable error) {
        log.error("websocket发生错误{}", error.toString());
    }

    /**
     * 往对应的全部连接发送消息
     *
     * @param vid  视频ID
     * @param text 消息内容，对象需转成JSON字符串
     */
    public void sendMessage(String vid, String text) {
        Set<Session> set = videoConnectionMap.get(vid);
        // 使用并行流往各客户端发送数据
        set.parallelStream().forEach(session -> {
            try {
                session.getBasicRemote().sendText(text);
            } catch (Exception e) {
                log.info("发送失败:{}", e.toString());
                throw new CustomException(AppHttpCodeEnum.SERVER_ERROR);
            }
        });
    }
}
